package com.shujia.wyh.jinjie;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.ArrayList;

public class HbaseAndHive {
    public static void main(String[] args) throws Exception{
        //创建表
        //获取配置文件对象
        Configuration conf = new Configuration();
        //2、设置以下zookeeper的集群信息
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        Connection conn = ConnectionFactory.createConnection(conf);
        Admin admin = conn.getAdmin();

        System.out.println("获取连接成功！！" + conn);
        System.out.println("HMaster的对象获取成功！！" + admin);
        System.out.println("------------------------------------------------------------------");

        TableName scores = TableName.valueOf("scores");

        //建表并读取文件插入到表中
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(scores);

        ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.of(Bytes.toBytes("info"));

        tableDescriptorBuilder.setColumnFamily(info);

        admin.createTable(tableDescriptorBuilder.build());

        System.out.println("表创建成功......");


        Table table = conn.getTable(scores);

        //创建一个List集合
        ArrayList<Put> puts = new ArrayList<>();

        //由于再hbase中相同的rowkey对应的列值重复添加会进行覆盖，所以我们要重新设计一下rowkey，让其不重复，数据也进行保留

        BufferedReader br = new BufferedReader(new FileReader("hbase/data/score.txt"));
        String line = null;
        while ((line=br.readLine())!=null){
            String[] split = line.split(",");
            String id = split[0];
            String sid = split[1];
            String score = split[2];

            Put put = new Put(Bytes.toBytes(id+"_"+sid));
            put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("score"),Bytes.toBytes(score));

            puts.add(put);
        }

        table.put(puts);

        System.out.println("数据添加完毕！！");


        /*

        select t1.id,t1.name,sum(cast(t2.score as int)) as sum_score from (select id,name from students_hbase where clazz='文科一班') t1 join (select split(id_sid,'_')[0] as id,split(id_sid,'_')[1] as sid,score from scores_hbase) t2 on(t1.id=t2.id) group by t1.id,t1.name order by sum_score desc limit 10;
         */




    }
}
